package com.cnebula.common.hdfs.http.impl;

import java.io.File;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.text.MessageFormat;

import org.apache.hadoop.security.authentication.client.AuthenticationException;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;
import org.osgi.service.component.ComponentContext;
import com.google.gson.Gson;

import com.cnebula.common.annotations.es.ESRef;
import com.cnebula.common.annotations.es.EasyService;
import com.cnebula.common.conf.IEasyServiceConfAdmin;
import com.cnebula.common.log.ILog;
import com.cnebula.common.hdfs.http.HadoopConf;
import com.cnebula.common.hdfs.http.URLUtil;
import com.cnebula.common.hdfs.http.WebHDFSConnection;

/**
 * 
 * ===== HTTP GET 
 * <li>OPEN 打开并读取一个文件
 * <li>GETFILESTATUS  获取文件/ 文件夹的状态信息
 * <li>LISTSTATUS  查看目录列表
 * <li>GETCONTENTSUMMARY 获取目录的上下文环境汇总信息
 * <li>GETFILECHECKSUM 获取文件校验和信息
 * <li>GETHOMEDIRECTORY 获取Home目录
 * <li>GETDELEGATIONTOKEN 获取代理Token
 * <li>GETDELEGATIONTOKENS 获取代理Tokens
 * ===== HTTP PUT 
 * <li>CREATE 创建并写一个文件
 * <li>MKDIRS 创建一个目录
 * <li>CREATESYMLINK 创建文件的符号链接
 * <li>RENAME 重名命文件、文件夹
 * <li>SETREPLICATION 设置备份
 * <li>SETOWNER 设置所有者
 * <li>SETPERMISSION 设置权限
 * <li>SETTIMES 设置文件修改时间
 * <li>RENEWDELEGATIONTOKEN 重新设置代理Token
 * <li>CANCELDELEGATIONTOKEN 取消代理Token
 * ===== HTTP POST 
 * APPEND 在文件末尾追加内容
 * ===== HTTP DELETE <br/>
 * DELETE 删除文件
 */

@EasyService
public class WebHDFSConnectionImpl implements WebHDFSConnection {

	@ESRef
	static ILog log;

	@ESRef
	IEasyServiceConfAdmin confAdmin;

	String httpfsUrl = "";
	String username = "";
	String superAdministrator = "";
	String user = "";

	protected void activate(ComponentContext ctx) throws Exception {
		HadoopConf conf = confAdmin.get(com.cnebula.common.hdfs.http.HadoopConf.class.getName(), HadoopConf.class);
		httpfsUrl = "http://" + conf.dfsHttpAddress;
		username = "user.name=" + conf.appCode;
		superAdministrator = "user.name="+conf.proxyuser;
		user = conf.appCode;
		
		if (httpfsUrl == null || httpfsUrl.trim().equals("")) {
			log.debug("WebHDFSConnection activate error, httpfsUrl is null",
					new Exception("WebHDFSConnection activate error, httpfsUrl is null"));
			return;
		}
		if (conf.appCode == null || conf.appCode.trim().equals("")) {
			log.error("WebHDFSConnection activate error, appCode is null",
					new Exception("WebHDFSConnection activate error, appCode is null"));
			return;
		}
		if (conf.proxyuser == null || conf.proxyuser.trim().equals("")) {
			log.error("WebHDFSConnection activate error, superAdministrator is null",
					new Exception("WebHDFSConnection activate error, superAdministrator is null"));
			return;
		}
		log.debug("WebHDFSConnection activate started, httpfsUrl=" + httpfsUrl);
	}

	protected static long copy(InputStream input, OutputStream result) throws IOException {
		byte[] buffer = new byte[12288]; // 8K=8192 12K=12288 64K=
		long count = 0L;
		int n = 0;
		while (-1 != (n = input.read(buffer))) {
			result.write(buffer, 0, n);
			count += n;
			result.flush();
		}
		result.flush();
		return count;
	}

	/**
	 * Report the result in JSON way
	 * 
	 * @param conn
	 * @param input
	 * @return
	 * @throws IOException
	 */
	private static String result(HttpURLConnection conn, boolean input) throws IOException {
		StringBuffer data = new StringBuffer();
		if (input) {
			InputStream is = conn.getInputStream();
			BufferedReader reader = new BufferedReader(new InputStreamReader(is, "utf-8"));
			String line = null;
			while ((line = reader.readLine()) != null) {
				log.debug(line);
				data.append(line);
			}
			reader.close();
			is.close();
		}
		Map<String, Object> result = new HashMap<String, Object>();
		result.put("code", conn.getResponseCode());
		result.put("mesg", conn.getResponseMessage());
		result.put("type", conn.getContentType());
		result.put("data", data);

		//
		// Convert a Map into JSON string.
		//
		Gson gson = new Gson();
		String json = gson.toJson(result);
		log.debug(json);

		return json;
	}

	/*
	 * ========================================================================
	 * GET
	 * ========================================================================
	 */
	/**
	 * <b>GETHOMEDIRECTORY</b>
	 * 
	 * curl -i "http://<HOST>:<PORT>/webhdfs/v1/?op=GETHOMEDIRECTORY"
	 * 
	 * @return
	 * @throws MalformedURLException
	 * @throws IOException
	 * @throws AuthenticationException
	 */
	public String getHomeDirectory() throws MalformedURLException, IOException, AuthenticationException {
		StringBuilder urlstr = new StringBuilder(httpfsUrl);
		urlstr.append("/webhdfs/v1/?op=GETHOMEDIRECTORY");
		urlstr.append("&").append(username);

		HttpURLConnection conn = (HttpURLConnection) new URL(urlstr.toString()).openConnection();
		conn.connect();

		String resp = result(conn, true);
		conn.disconnect();
		return resp;
	}

	/**
	 * <b>OPEN</b>
	 * 
	 * curl -i -L "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=OPEN [&offset=
	 * <LONG>][&length=<LONG>][&buffersize=<INT>]"
	 * 
	 * @param path
	 * @param os
	 * @throws AuthenticationException
	 * @throws IOException
	 * @throws MalformedURLException
	 */
	public String open(String path, OutputStream os)
			throws MalformedURLException, IOException, AuthenticationException {

		StringBuilder urlstr = new StringBuilder(httpfsUrl);
		urlstr.append(MessageFormat.format("/webhdfs/v1/{0}?op=OPEN", URLUtil.encodePath(path)));
		urlstr.append("&").append(username);

		HttpURLConnection conn = (HttpURLConnection) new URL(urlstr.toString()).openConnection();
		conn.setRequestMethod("GET");
		conn.setRequestProperty("Content-Type", "application/octet-stream");
		conn.connect();
		InputStream is = conn.getInputStream();
		copy(is, os);
		is.close();
		os.close();
		String resp = result(conn, false);
		conn.disconnect();
		
		return resp;
	}

	/**
	 * <b>GETCONTENTSUMMARY</b>
	 * 
	 * curl -i "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=GETCONTENTSUMMARY"
	 * 
	 * @param path
	 * @return
	 * @throws MalformedURLException
	 * @throws IOException
	 * @throws AuthenticationException
	 */
	public String getContentSummary(String path) throws MalformedURLException, IOException, AuthenticationException {

		StringBuilder urlstr = new StringBuilder(httpfsUrl);
		urlstr.append(MessageFormat.format("/webhdfs/v1/{0}?op=GETCONTENTSUMMARY", URLUtil.encodePath(path)));
		urlstr.append("&").append(username);

		HttpURLConnection conn = (HttpURLConnection) new URL(urlstr.toString()).openConnection();
		conn.setRequestMethod("GET");
		conn.connect();
		String resp = result(conn, true);
		conn.disconnect();

		return resp;
	}

	/**
	 * <b>LISTSTATUS</b>
	 * 
	 * curl -i "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=LISTSTATUS"
	 * 
	 * @param path
	 * @return
	 * @throws MalformedURLException
	 * @throws IOException
	 * @throws AuthenticationException
	 */
	public String listStatus(String path) throws MalformedURLException, IOException, AuthenticationException {
		StringBuilder urlstr = new StringBuilder(httpfsUrl);
		urlstr.append(MessageFormat.format("/webhdfs/v1/{0}?op=LISTSTATUS", URLUtil.encodePath(path)));
		urlstr.append("&").append(username);

		HttpURLConnection conn = (HttpURLConnection) new URL(urlstr.toString()).openConnection();
		conn.setRequestMethod("GET");
		conn.connect();
		String resp = result(conn, true);
		conn.disconnect();

		return resp;
	}

	/**
	 * <b>GETFILESTATUS</b>
	 * 
	 * curl -i "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=GETFILESTATUS"
	 * 
	 * @param path
	 * @return
	 * @throws MalformedURLException
	 * @throws IOException
	 * @throws AuthenticationException
	 */
	public String getFileStatus(String path) throws MalformedURLException, IOException, AuthenticationException {
		StringBuilder urlstr = new StringBuilder(httpfsUrl);
		urlstr.append(MessageFormat.format("/webhdfs/v1/{0}?op=GETFILESTATUS", URLUtil.encodePath(path)));
		urlstr.append("&").append(username);

		HttpURLConnection conn = (HttpURLConnection) new URL(urlstr.toString()).openConnection();
		conn.setRequestMethod("GET");
		conn.connect();
		String resp = result(conn, true);
		conn.disconnect();

		return resp;
	}

	/**
	 * <b>GETFILECHECKSUM</b>
	 * 
	 * curl -i "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=GETFILECHECKSUM"
	 * 
	 * @param path
	 * @return
	 * @throws MalformedURLException
	 * @throws IOException
	 * @throws AuthenticationException
	 */
	public String getFileCheckSum(String path) throws MalformedURLException, IOException, AuthenticationException {
		String resp = null;
		StringBuilder urlstr = new StringBuilder(httpfsUrl);
		urlstr.append(MessageFormat.format("/webhdfs/v1/{0}?op=GETFILECHECKSUM", URLUtil.encodePath(path)));
		urlstr.append("&").append(username);
		HttpURLConnection conn = (HttpURLConnection) new URL(urlstr.toString()).openConnection();
		conn.setRequestMethod("GET");
		conn.connect();
		resp = result(conn, true);
		conn.disconnect();

		return resp;
	}

	/*
	 * ========================================================================
	 * PUT
	 * ========================================================================
	 */
	/**
	 * <b>CREATE</b>
	 * 
	 * curl -i -X PUT "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=CREATE
	 * [&overwrite=<true|false>][&blocksize=<LONG>][&replication=<SHORT>]
	 * [&permission=<OCTAL>][&buffersize=<INT>]"
	 * 
	 * @param path
	 * @param is
	 * @return
	 * @throws MalformedURLException
	 * @throws IOException
	 * @throws AuthenticationException
	 */
	public String create(String path, InputStream is)
			throws MalformedURLException, IOException, AuthenticationException {
		String resp = null;
		String redirectUrl = null;
		StringBuilder urlstr = new StringBuilder(httpfsUrl);
		urlstr.append(MessageFormat.format("/webhdfs/v1/{0}?op=CREATE", URLUtil.encodePath(path)));
		urlstr.append("&").append(username);
		HttpURLConnection conn = (HttpURLConnection) new URL(urlstr.toString()).openConnection();
		conn.setRequestMethod("PUT");
		conn.setInstanceFollowRedirects(false);
		conn.connect();
		log.info("Location:" + conn.getHeaderField("Location"));
		resp = result(conn, true);
		if (conn.getResponseCode() == 307)
			redirectUrl = conn.getHeaderField("Location");
		conn.disconnect();

		if (redirectUrl != null) {
			conn = (HttpURLConnection) new URL(redirectUrl).openConnection();
			conn.setRequestMethod("PUT");
			conn.setDoOutput(true);
			conn.setDoInput(true);
			conn.setUseCaches(false);
			conn.setRequestProperty("Content-Type", "application/octet-stream");
			// conn.setRequestProperty("Transfer-Encoding", "chunked");
			final int _SIZE = is.available();
			conn.setRequestProperty("Content-Length", "" + _SIZE);
			conn.setFixedLengthStreamingMode(_SIZE);
			conn.connect();
			OutputStream os = conn.getOutputStream();
			copy(is, os);
			// Util.copyStream(is, os);
			is.close();
			os.close();
			resp = result(conn, false);
			log.info(resp);
			conn.disconnect();
		}

		return resp;
	}

	/*
	 * ========================================================================
	 * PUT
	 * ========================================================================
	 */
	/**
	 * <b>CREATE</b>
	 * 
	 * curl -i -X PUT "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=CREATE
	 * [&overwrite=<true|false>][&blocksize=<LONG>][&replication=<SHORT>]
	 * [&permission=<OCTAL>][&buffersize=<INT>]"
	 * 
	 * @param path
	 * @param is
	 * @return
	 * @throws MalformedURLException
	 * @throws IOException
	 * @throws AuthenticationException
	 */
	public String createPOC(String path, InputStream is, HashMap<String, String> map)
			throws MalformedURLException, IOException, AuthenticationException {
		String resp = null;
		String redirectUrl = null;
		StringBuilder urlstr = new StringBuilder(httpfsUrl);
		urlstr.append(MessageFormat.format("/webhdfs/v1/{0}?op=CREATE", URLUtil.encodePath(path)));
		urlstr.append("&").append(username);
		HttpURLConnection conn = (HttpURLConnection) new URL(urlstr.toString()).openConnection();
		conn.setRequestMethod("PUT");
		conn.setInstanceFollowRedirects(false);
		conn.connect();
		log.debug("Location:" + conn.getHeaderField("Location"));
		resp = result(conn, true);
		if (conn.getResponseCode() == 307) {
			String str = conn.getHeaderField("Location");
			if (str.startsWith("http://big1.big")) {
				redirectUrl = str.replaceAll("big1.big", map.get("big1.big"));
			} else if (str.startsWith("http://big2.big")) {
				redirectUrl = str.replaceAll("big2.big", map.get("big2.big"));
			} else if (str.startsWith("http://big3.big")) {
				redirectUrl = str.replaceAll("big3.big", map.get("big3.big"));
			} else {
				redirectUrl = conn.getHeaderField("Location");
			}
		}
		conn.disconnect();

		if (redirectUrl != null) {
			conn = (HttpURLConnection) new URL(redirectUrl).openConnection();
			conn.setRequestMethod("PUT");
			conn.setDoOutput(true);
			conn.setDoInput(true);
			conn.setUseCaches(false);
			conn.setRequestProperty("Content-Type", "application/octet-stream");
			final int _SIZE = is.available();
			conn.setRequestProperty("Content-Length", "" + _SIZE);
			conn.setFixedLengthStreamingMode(_SIZE);
			conn.connect();
			OutputStream os = conn.getOutputStream();
			copy(is, os);
			is.close();
			os.close();
			resp = result(conn, false);
			conn.disconnect();
		}

		return resp;
	}

	/**
	 * <b>MKDIRS</b>
	 * 
	 * curl -i -X PUT
	 * "http://<HOST>:<PORT>/<PATH>?op=MKDIRS[&permission=<OCTAL>]"
	 * 
	 * @param path
	 * @return
	 * @throws AuthenticationException
	 * @throws IOException
	 * @throws MalformedURLException
	 */
	public String mkdirs1(String path) throws MalformedURLException, IOException, AuthenticationException {
		String resp = null;
		StringBuilder urlstr = new StringBuilder(httpfsUrl);
		urlstr.append(MessageFormat.format("/webhdfs/v1/{0}?op=MKDIRS", URLUtil.encodePath(path)));
		urlstr.append("&").append(username);
		HttpURLConnection conn = (HttpURLConnection) new URL(urlstr.toString()).openConnection();
		conn.setRequestMethod("PUT");
		conn.connect();
		resp = result(conn, true);
		conn.disconnect();
		return resp;
	}
	
	/**
	 * mktopdirs 创建顶级目录
	 * 
	 * curl -i -X PUT
	 * "http://<HOST>:<PORT>/<PATH>?op=MKDIRS[&permission=<OCTAL>]"
	 * 
	 * @param path
	 * @return
	 * @throws AuthenticationException
	 * @throws IOException
	 * @throws MalformedURLException
	 * @throws JSONException 
	 */
	public String mkdirs(String path) throws MalformedURLException, IOException, AuthenticationException {
		String result = "";
		String topFileName="";
		String fileseparator =String.valueOf(File.separator);
		if(path.contains(fileseparator)){
			String[] pathList = path.split(File.separator);
			topFileName = pathList[0].trim();
		}else{
			topFileName = path.trim();
		}
		
		if(!checkFileName(topFileName)  && !checkOwner(topFileName)){ //顶级目录文件名存在，并且该文件的owner不是本user
			log.info("文件已经存在！请重新命名文件");
			//做异常处理
		}else if(!checkFileName(topFileName)  && checkOwner(topFileName)){ //顶级目录文件名存在，并且就是本user创建的
			result = mkdirs1(path);
		}else{          //顶级目录文件名不存在，直接创建
			mkmydirs(topFileName);
			setOwner(topFileName);
			result = mkdirs1(path);
		}
		return result;
	}
	
	
	/**
	 * checkFileName 校验文件名称是否重复
	 * 
	 * @param path
	 * @return
	 * @throws AuthenticationException
	 * @throws IOException
	 * @throws MalformedURLException
	 * @throws JSONException 
	 */
	public boolean checkFileName(String topFileName)  throws MalformedURLException, IOException, AuthenticationException{
			String str = listStatus(File.separator);
			ArrayList<String> fileNames = new ArrayList<String>();
			try {
				JSONObject  json= new JSONObject(str);
				String data =  (String) json.get("data");
				String dataStr =data .replaceAll("\\\\","");
				JSONObject  FileStatuses= new JSONObject(dataStr);
				JSONObject FileStatus = (JSONObject) FileStatuses.get("FileStatuses");
			    JSONArray jsonArray = (JSONArray) FileStatus.get("FileStatus");
				for(int j=0;j<jsonArray.length();j++){
					JSONObject fileinfo = (JSONObject) jsonArray.get(j);
					String filename = (String) fileinfo.get("pathSuffix");
					fileNames.add(filename);
				}
			} catch (JSONException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			for(int k =0;k<fileNames.size();k++){
				if(topFileName.equals(fileNames.get(k))){
					return false;
				}
			}
		return true;
	}
	
	public boolean checkOwner(String topFileName)  throws MalformedURLException, IOException, AuthenticationException{
		String fileStatus = getFileStatus(topFileName);
		String owner ="";
		try {
			JSONObject  json= new JSONObject(fileStatus);
			String data =  (String) json.get("data");
			String dataStr =data .replaceAll("\\\\","");
			JSONObject  dataObject= new JSONObject(dataStr);
			JSONObject status = (JSONObject) dataObject.get("FileStatus");
			owner = (String) status.get("owner");
		} catch (JSONException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		if(!owner.equals(user)){
			return false;
		}
		return true;
	}
	
	public String mkmydirs(String path) throws MalformedURLException, IOException, AuthenticationException {
		String resp = null;
		StringBuilder urlstr = new StringBuilder(httpfsUrl);
		urlstr.append(MessageFormat.format("/webhdfs/v1/{0}?op=MKDIRS", URLUtil.encodePath(path)));
		urlstr.append("&").append(superAdministrator);
		HttpURLConnection conn = (HttpURLConnection) new URL(urlstr.toString()).openConnection();
		conn.setRequestMethod("PUT");
		conn.connect();
		resp = result(conn, true);
		conn.disconnect();
		return resp;
	}
	

	/**
	 * <b>CREATESYMLINK</b>
	 * 
	 * curl -i -X PUT "http://<HOST>:<PORT>/<PATH>?op=CREATESYMLINK
	 * &destination=<PATH>[&createParent=<true|false>]"
	 * 
	 * @param path
	 * @return
	 * @throws AuthenticationException
	 * @throws IOException
	 * @throws MalformedURLException
	 */
	public String createSymLink(String srcPath, String destPath)
			throws MalformedURLException, IOException, AuthenticationException {
		String resp = null;
		StringBuilder urlstr = new StringBuilder(httpfsUrl);
		urlstr.append(MessageFormat.format("/webhdfs/v1/{0}?op=CREATESYMLINK&destination={1}",
				URLUtil.encodePath(srcPath), URLUtil.encodePath(destPath)));
		urlstr.append("&").append(username);
		HttpURLConnection conn = (HttpURLConnection) new URL(urlstr.toString()).openConnection();
		conn.setRequestMethod("PUT");
		conn.connect();
		resp = result(conn, true);
		conn.disconnect();

		return resp;
	}

	/**
	 * <b>RENAME</b>
	 * 
	 * curl -i -X PUT "http://<HOST>:<PORT>/<PATH>?op=RENAME &destination=
	 * <PATH>[&createParent=<true|false>]"
	 * 
	 * @param path
	 * @return
	 * @throws AuthenticationException
	 * @throws IOException
	 * @throws MalformedURLException
	 */
	public String rename(String srcPath, String destPath)
			throws MalformedURLException, IOException, AuthenticationException {
		String resp = null;
		StringBuilder urlstr = new StringBuilder(httpfsUrl);
		urlstr.append(MessageFormat.format("/webhdfs/v1/{0}?op=RENAME&destination={1}", URLUtil.encodePath(srcPath),
				URLUtil.encodePath(destPath)));
		urlstr.append("&").append(username);
		HttpURLConnection conn = (HttpURLConnection) new URL(urlstr.toString()).openConnection();
		conn.setRequestMethod("PUT");
		conn.connect();
		resp = result(conn, true);
		conn.disconnect();

		return resp;
	}

	/**
	 * SETPERMISSION
	 * 
	 * curl -i -X PUT "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=SETPERMISSION
	 * [&permission=<OCTAL>]"
	 * 
	 * @param path
	 * @return
	 * @throws AuthenticationException
	 * @throws IOException
	 * @throws MalformedURLException
	 */
	public String setPermission(String path) throws MalformedURLException, IOException, AuthenticationException {
		String resp = null;
		StringBuilder urlstr = new StringBuilder(httpfsUrl);
		urlstr.append(MessageFormat.format("/webhdfs/v1/{0}?op=SETPERMISSION", URLUtil.encodePath(path)));
		urlstr.append("&").append(username);
		HttpURLConnection conn = (HttpURLConnection) new URL(urlstr.toString()).openConnection();
		conn.setRequestMethod("PUT");
		conn.connect();
		resp = result(conn, true);
		conn.disconnect();

		return resp;
	}

	/**
	 * <b>SETOWNER</b>
	 * 
	 * curl -i -X PUT "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=SETOWNER
	 * [&owner=<USER>][&group=<GROUP>]"
	 * 
	 * @param path
	 * @return
	 * @throws AuthenticationException
	 * @throws IOException
	 * @throws MalformedURLException
	 */
	public String setOwner(String path) throws MalformedURLException, IOException, AuthenticationException {
		String resp = null;
		StringBuilder urlstr = new StringBuilder(httpfsUrl);
		urlstr.append(MessageFormat.format("/webhdfs/v1/{0}?op=SETOWNER", URLUtil.encodePath(path)));
		urlstr.append("&").append("owner=" +user);
		urlstr.append("&").append(superAdministrator);
		HttpURLConnection conn = (HttpURLConnection) new URL(urlstr.toString()).openConnection();
		conn.setRequestMethod("PUT");
		conn.connect();
		resp = result(conn, true);
		conn.disconnect();

		return resp;
	}
	

	/**
	 * <b>SETREPLICATION</b>
	 * 
	 * curl -i -X PUT "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=SETREPLICATION
	 * [&replication=<SHORT>]"
	 * 
	 * @param path
	 * @return
	 * @throws AuthenticationException
	 * @throws IOException
	 * @throws MalformedURLException
	 */
	public String setReplication(String path) throws MalformedURLException, IOException, AuthenticationException {
		String resp = null;
		StringBuilder urlstr = new StringBuilder(httpfsUrl);
		urlstr.append(MessageFormat.format("/webhdfs/v1/{0}?op=SETREPLICATION", URLUtil.encodePath(path)));
		urlstr.append("&").append(username);
		HttpURLConnection conn = (HttpURLConnection) new URL(urlstr.toString()).openConnection();
		conn.setRequestMethod("PUT");
		conn.connect();
		resp = result(conn, true);
		conn.disconnect();

		return resp;
	}

	/**
	 * <b>SETTIMES</b>
	 * 
	 * curl -i -X PUT "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=SETTIMES
	 * [&modificationtime=<TIME>][&accesstime=<TIME>]"
	 * 
	 * @param path
	 * @return
	 * @throws AuthenticationException
	 * @throws IOException
	 * @throws MalformedURLException
	 */
	public String setTimes(String path) throws MalformedURLException, IOException, AuthenticationException {
		String resp = null;
		StringBuilder urlstr = new StringBuilder(httpfsUrl);
		urlstr.append(MessageFormat.format("/webhdfs/v1/{0}?op=SETTIMES", URLUtil.encodePath(path)));
		urlstr.append("&").append(username);
		HttpURLConnection conn = (HttpURLConnection) new URL(urlstr.toString()).openConnection();
		conn.setRequestMethod("PUT");
		conn.connect();
		resp = result(conn, true);
		conn.disconnect();

		return resp;
	}

	/*
	 * ========================================================================
	 * POST
	 * ========================================================================
	 */
	/**
	 * curl -i -X POST
	 * "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=APPEND[&buffersize=<INT>]"
	 * 
	 * @param path
	 * @param is
	 * @return
	 * @throws MalformedURLException
	 * @throws IOException
	 * @throws AuthenticationException
	 */
	public String append(String path, InputStream is)
			throws MalformedURLException, IOException, AuthenticationException {
		String resp = null;
		String redirectUrl = null;
		StringBuilder urlstr = new StringBuilder(httpfsUrl);
		urlstr.append(MessageFormat.format("/webhdfs/v1/{0}?op=APPEND", URLUtil.encodePath(path)));
		urlstr.append("&").append(username);

		HttpURLConnection conn = (HttpURLConnection) new URL(urlstr.toString()).openConnection();
		conn.setRequestMethod("POST");
		conn.setInstanceFollowRedirects(false);
		conn.connect();
		log.debug("Location:" + conn.getHeaderField("Location"));
		resp = result(conn, true);
		if (conn.getResponseCode() == 307)
			redirectUrl = conn.getHeaderField("Location");
		conn.disconnect();

		if (redirectUrl != null) {
			conn = (HttpURLConnection) new URL(redirectUrl).openConnection();
			conn.setRequestMethod("POST");
			conn.setDoOutput(true);
			conn.setDoInput(true);
			conn.setUseCaches(false);
			conn.setRequestProperty("Content-Type", "application/octet-stream");
			// conn.setRequestProperty("Transfer-Encoding", "chunked");
			final int _SIZE = is.available();
			conn.setRequestProperty("Content-Length", "" + _SIZE);
			conn.setFixedLengthStreamingMode(_SIZE);
			conn.connect();
			OutputStream os = conn.getOutputStream();
			copy(is, os);
			// Util.copyStream(is, os);
			is.close();
			os.close();
			resp = result(conn, true);
			conn.disconnect();
		}

		return resp;
	}

	/*
	 * ========================================================================
	 * DELETE
	 * ========================================================================
	 */
	/**
	 * <b>DELETE</b>
	 * 
	 * curl -i -X DELETE "http://<host>:<port>/webhdfs/v1/<path>?op=DELETE
	 * [&recursive=<true|false>]"
	 * 
	 * @param path
	 * @return
	 * @throws AuthenticationException
	 * @throws IOException
	 * @throws MalformedURLException
	 */
	public String delete(String path) throws MalformedURLException, IOException, AuthenticationException {
		String resp = null;

		StringBuilder urlstr = new StringBuilder(httpfsUrl);
		urlstr.append(MessageFormat.format("/webhdfs/v1/{0}?op=DELETE", URLUtil.encodePath(path)));
		urlstr.append("&").append(username);

		HttpURLConnection conn = (HttpURLConnection) new URL(urlstr.toString()).openConnection();
		conn.setRequestMethod("DELETE");
		conn.setInstanceFollowRedirects(false);
		conn.connect();
		resp = result(conn, true);
		conn.disconnect();

		return resp;
	}
}